package Redis

import (
	"encoding/json"
	"fmt"
	"gitee.com/xixigroup/sweet-tools/base/Utils"
	"github.com/gomodule/redigo/redis"
	"log"
	"time"
)

/*
	基于redis发布订阅模式实现的消息总线，使用原则：全局应且只应有一个数据总线
	Usage:
		func test(message *redis.Message) (bool, error) {
			fmt.Println("这是test", message)
			return true, nil
		}

		event := redis.CreateDistributedEvent(Pool, "channel", 3) // Pool redigo *redis.Pool
		message := redis.Message{
			Msg: "消息来了",
			Tag: "test_tag",
		}
		time.Sleep(time.Second)
		event.AddListener(message.Tag, test)
		event.Dispatch(&message)
		time.Sleep(time.Second * 10)
	output:
		2021/08/04 17:59:37 Event listener start channel: EVENT_BUS_3298190398_0
		2021/08/04 17:59:37 Event listener start channel: EVENT_BUS_3298190398_1
		2021/08/04 17:59:37 Event listener start channel: EVENT_BUS_3298190398_2
		2021/08/04 17:59:38 Event receive message: {"msg":"消息来了","tag":"test_tag"}
		这是test &{消息来了 test_tag}

*/

var Event *DistributedEvent

type DistributedEvent struct {
	/*
		Pool redigo *redis.Pool
		channels // channels
		tasks  tag-callback
	*/
	Pool     *redis.Pool
	channels *[]string
	tasks    map[string]*Task // tag-callback
}

type Task struct {
	tag      string
	function func(message *Message) (bool, error)
}

func CreateDistributedEvent(
	// 创建数据总线
	pool *redis.Pool, channelName string, channelCount int) *DistributedEvent {

	event := DistributedEvent{}
	event.Pool = pool
	event.tasks = make(map[string]*Task)
	channelNameHash := Utils.HashMath(channelName)
	event.channels = &[]string{}
	for i := 0; i < channelCount; i++ {
		channel := fmt.Sprintf("EVENT_BUS_%d_%d", channelNameHash, i)
		*event.channels = append(*event.channels, channel)
	}

	Event = &event
	for _, channel := range *event.channels {
		go event.run(channel)
	}
	return &event
}

func (event *DistributedEvent) GetChannel(channel string) string {
	// 格式化channel
	return fmt.Sprintf("event_bus_%s", channel)
}

type Message struct {
	Msg string `json:"msg"`
	Tag string `json:"tag"`
}

func (event *DistributedEvent) Dispatch(message *Message) (bool, error) {
	// 发布消息
	defer Utils.TracebackFormatExec(fmt.Sprintf("Event Dispatch [%v] error", *message))

	conn := event.Pool.Get()
	defer conn.Close()
	channel := (*event.channels)[int(Utils.HashMath(message.Tag))%len(*event.channels)]

	bytesMsg, _ := json.Marshal(message)
	res, err := conn.Do("PUBLISH", channel, string(bytesMsg))
	if err != nil || res == nil {
		return false, err
	}
	row, err := Utils.AssertInt(res)
	if err != nil {
		return false, err
	}
	if row == 1 {
		return true, nil
	}
	return false, err
}

func (event *DistributedEvent) AddListener(tag string, callback func(message *Message) (bool, error)) {
	// 添加监听channel
	task := Task{tag, callback}
	event.tasks[tag] = &task
}

func (event *DistributedEvent) RemoveListener(tag string) {
	// 移除监听channel
	if _, ok := event.tasks[tag]; ok {
		delete(event.tasks, tag)
	}
}

func (event *DistributedEvent) run(channel string) {
	// 监听上层调用，这里for死循环好处在于如果下层因网络原因导致redis连接失败，这里有机会重置连接
	log.Println("Event listener start channel:", channel)

	for {
		event.listening(channel)
		time.Sleep(time.Second)
	}
}

func (event *DistributedEvent) listening(channel string) {
	// 监听消息主逻辑
	defer Utils.TracebackFormatExec(fmt.Sprintf("Event listening[%s] error", channel))

	conn := event.Pool.Get()
	defer conn.Close()

	psc := redis.PubSubConn{Conn: conn}

	if err := psc.Subscribe(channel); err != nil {
		log.Println("Event listening error: ", err)
		return
	}

	for {
		msg := psc.Receive()
		switch n := msg.(type) {

		case redis.Message:
			var msg = Message{}
			message := string(n.Data)
			json.Unmarshal(n.Data, &msg)

			if msg.Tag == "" {
				log.Printf("Event receive invalid message[%s]", message)
			}
			if f, ok := event.tasks[msg.Tag]; ok {
				log.Printf("Event receive message[%v]", msg)
				go f.function(&msg)
			}

		case redis.Subscription:
			continue

		default:
			return
		}
	}
}
